package com.neo.duan.rxjava;

import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.util.Log;
import android.view.View;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.Scheduler;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.SchedulerSupport;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.internal.schedulers.IoScheduler;
import io.reactivex.internal.schedulers.NewThreadScheduler;
import io.reactivex.schedulers.Schedulers;

/**
 * @author neo.duan
 * @date 2017/12/19 15:02
 * @desc 请描述该文件
 */
public class MainActivity extends AppCompatActivity {
    private final String TAG = "MainActivity";

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
    }

    public void onClickTest(View view) {
        try {
            test9();
        } catch (Exception e) {
            log(e.getMessage());
        }
    }

    /**
     * 普通调用
     */
    private void test1() {
        //1.创建被观察者
        Observable observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter e) throws Exception {
                //do something
                e.onNext("subscribe  哈哈");
                e.onComplete();
            }
        });

        // 2.创建观察者
        Observer<String> observer = new MyObserver<>();

        //订阅
        observable.subscribe(observer);
    }

    /**
     * 链式调用
     */
    private void test2() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {

            }
        }).subscribe(new MyObserver<String>());
    }

    /**
     * just简化
     */
    public void test3() {
        Observable.just("hello", "world", "abc").subscribe(new MyObserver<String>());
    }

    private void test4() {
        String[] arr = {"a", "b", "c"};
        Observable.fromArray(arr).subscribe(new MyObserver<String>());
    }

    /**
     * Consumer只关心onNext((
     */
    private void test5() {
        Observable.just("hello", "world").subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                log(s);
            }
        });
    }

    private void test6() {
        log("main thread:" + Thread.currentThread().getName());

        Observable observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter e) throws Exception {
                //do something
                e.onNext("subscribe  哈哈");
                log("subscribe thread:" + Thread.currentThread().getName());
                e.onComplete();
            }
        });

        //上游发送事件线程
        observable.subscribeOn(Schedulers.newThread())
                //下游接收事件线程
                .observeOn(Schedulers.newThread())
                .doOnNext(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        log("doOnNext accept: " + s);
                    }
                })
                .map(new Function<Integer, String>() {

                    //接收的参数Integer
                    @Override
                    public String apply(Integer str) throws Exception {
//                        new Function<Integer, String>
                        //Integer:定义传入参数
                        //String：定义返回参数
                        return "This is result :" + str;
                    }
                })
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        log(s);
                        log("accept thread:" + Thread.currentThread().getName());
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        log("accept error");
                    }
                });
    }

    /**
     * flatMap
     */
    private void test7() {
        Observable.create(new ObservableOnSubscribe() {
            @Override
            public void subscribe(ObservableEmitter e) throws Exception {
                e.onNext(5);
                e.onNext(6);
                e.onNext(7);
            }
        }).concatMap(new Function<Integer, String>() {

            @Override
            public String apply(Integer integer) throws Exception {
                List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("" + (i + integer));
                }
                return Observable.fromIterable(list).delay(5, TimeUnit.SECONDS).blockingFirst();
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                log("accept : " + s);
            }
        });
    }

    /**
     * zip使用
     */
    private void test8() {
        Observable observable1 = Observable.create(new ObservableOnSubscribe() {
            @Override
            public void subscribe(ObservableEmitter e) throws Exception {
                e.onNext("1");
                log("onNext1");
                e.onNext("2");
                log("onNext2");
                e.onNext("3");
                log("onNext3");
            }
        }).observeOn(Schedulers.newThread());

        Observable observable2 = Observable.create(new ObservableOnSubscribe() {
            @Override
            public void subscribe(ObservableEmitter e) throws Exception {
                e.onNext("a");
                log("onNexta");
                e.onNext("b");
                log("onNextb");
                e.onNext("c");
                log("onNextc");
            }
        }).observeOn(AndroidSchedulers.mainThread());

        Observable.zip(observable1, observable2, new BiFunction<String, String, String>() {
            @Override
            public String apply(String str1, String str2) throws Exception {
                return str1 + str2;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String str) throws Exception {
                log(str);
            }
        });
    }

    /**
     * filter
     */
    private void  test9() {
        Observable.just("1", "2", "3")
                .filter(new Predicate<String>() {
                    @Override
                    public boolean test(String s) throws Exception {
                        return !s.equals("2");
                    }
                }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                log("accept:" + s);
            }
        });
    }


    class MyObserver<T> implements Observer<T> {

        @Override
        public void onSubscribe(Disposable d) {
            log("onSubscribe ");
            d.dispose();
        }

        @Override
        public void onNext(T o) {
            log("onNext " + o.toString());
        }

        @Override
        public void onError(Throwable e) {
            log("onError ");
        }

        @Override
        public void onComplete() {
            log("onComplete ");
        }
    }

    private void log(String log) {
        Log.d(TAG, log);
    }
}
